
#include "./kv_server.h"
#include "../util/lutils.h"

namespace lukv {

kv_server::kv_server() : kv_server(4399, "127.0.0.1") { }

kv_server::kv_server(int _p, std::string _h) : port(_p), host(std::move(_h)) { }

kv_server::~kv_server() = default;

#ifdef __OSX__
int kv_server::start() {
    int BACKLOG_SIZE = 16;
    int READ_BUF_SIZE = 2048;
    int K_EVENTS_CNT = 128;

    // socket
    int srv_sock = socket(AF_INET, SOCK_STREAM, 0);
    fcntl(srv_sock, F_SETFL, O_NONBLOCK | fcntl(srv_sock, F_GETFL, 0));

    // address
    sockaddr_in srv_addr { };
    memset(&srv_addr, 0, sizeof(srv_addr));
    srv_addr.sin_port = htons(port);
    srv_addr.sin_addr.s_addr = inet_addr(host.c_str());
    srv_addr.sin_family = AF_INET;

    // bind
    if (bind(srv_sock, (sockaddr *)&srv_addr, sizeof(srv_addr)) < 0) {
        close(srv_sock);
        EXIT_ERR(strerror(errno));
    }

    // listen
    if (listen(srv_sock, BACKLOG_SIZE) < 0) {
        close(srv_sock);
        EXIT_ERR(strerror(errno));
    }

    // kqueue
    int kq_fd = kqueue();

    // events to monitor
    struct kevent k_read_event { (uint64_t) srv_sock, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, nullptr };

    // add events to k queue
    if (kevent(kq_fd, &k_read_event, 1, nullptr, 0, nullptr) < 0) {
        close_fds(srv_sock, kq_fd);
        EXIT_ERR(strerror(errno))
    }

    // events to handle
    struct kevent out_events[K_EVENTS_CNT];

    // main loop
    char read_buf[READ_BUF_SIZE];
    while (true) {
        // wait for events to occur
        int cnt = kevent(kq_fd, nullptr, 0, out_events, K_EVENTS_CNT, nullptr);

        if (cnt < 0) {
            LOG_ERR(strerror(errno));
            break;
        }
        else {
            // handle events foreach
            for (int i = 0; i < cnt; ++i) {
                auto _sock = out_events[i].ident;
                auto _data = out_events[i].data;
                // handle READ event
                if (out_events[i].filter & EVFILT_READ) {
                    std::cout << "HANDLE READ EVENT" << "[ident]" << _sock << "[data]" << _data << std::endl;
                    if (_sock == srv_sock) {
                        sockaddr_in cli_addr { };
                        socklen_t cli_addr_len = sizeof(cli_addr);
                        int cli_sock = -1;
                        for (int j = 0; j < _data; ++j) {
                            cli_sock = accept(_sock, (sockaddr *) &cli_addr, &cli_addr_len);

                            if (cli_sock < 0) {
                                LOG_ERR(strerror(errno));
                                continue;
                            }

                            std::cout << "[sock]" << cli_sock << "[client]" << inet_ntoa(cli_addr.sin_addr) << ":" << ntohs(cli_addr.sin_port) << std::endl;

                            auto rlen = read(cli_sock, read_buf, READ_BUF_SIZE);
                            std::string read_str(read_buf, rlen);

                            std::cout << "[read]<" << read_str << ">" << std::endl;

                            std::pair<char, std::string> _ret = base.handle_f(read_str);
                            std::string wt_msg;
                            wt_msg.push_back(_ret.first >= 0 ? 'S' : 'E');
                            wt_msg.append(_ret.second);

                            write(cli_sock, wt_msg.c_str(), wt_msg.size());

                            std::cout << "[write]<" << wt_msg << ">" << std::endl;

                            close(cli_sock);
                        }
                    }
                    else {
                        auto rlen = recv(_sock, read_buf, READ_BUF_SIZE, 0);
                        std::string read_str(read_buf, rlen);

                        std::cout << "[recv]<" << read_str << ">" << std::endl;
                    }
                }
            }
        }
    }

    close_fds(srv_sock, kq_fd);

    return 0;
}
#endif

#ifdef __LINUX__
int kv_server::start() {

    const int EP_EVENT_CNT = 16;
    const int BACKLOG_SIZE = 32;
    const int READ_BUF_SIZE = 1024;

    // socket
    int srv_sock = socket(AF_INET, SOCK_STREAM, 0);
    fcntl(srv_sock, F_SETFL, O_NONBLOCK | fcntl(srv_sock, F_GETFL, 0));

    // address
    sockaddr_in srv_addr { };
    memset(&srv_addr, 0, sizeof(srv_addr));
    srv_addr.sin_port = htons(port);
    srv_addr.sin_addr.s_addr = inet_addr(host.c_str());
    srv_addr.sin_family = AF_INET;

    // bind
    if (bind(srv_sock, (sockaddr *)&srv_addr, sizeof(srv_addr)) < 0) {
        close(srv_sock);
        EXIT_ERR(strerror(errno));
    }

    // listen
    if (listen(srv_sock, BACKLOG_SIZE) < 0) {
        close(srv_sock);
        EXIT_ERR(strerror(errno));
    }

    int ep_fd = epoll_create(64);

    struct epoll_event srv_event { };
    srv_event.events = EPOLLIN;
    srv_event.data.fd = srv_sock;

    epoll_ctl(ep_fd, EPOLL_CTL_ADD, srv_sock, &srv_event);

    struct epoll_event ep_events[EP_EVENT_CNT];

    char read_buf[READ_BUF_SIZE];

    while (true) {
        int cnt = epoll_wait(ep_fd, ep_events, EP_EVENT_CNT, 500);
        if (cnt < 0) {
            LOG_ERR(strerror(errno));
            break;
        }
        
        for (int i = 0; i < cnt; i++) {
            int _sock = ep_events[i].data.fd;

            if (_sock == srv_sock) {
                struct sockaddr_in cli_addr { };
                socklen_t cli_addr_len = sizeof(cli_addr);

                int cli_sock = accept(srv_sock, (sockaddr *)&cli_addr, &cli_addr_len);
                if (cli_sock < 0) {
                    LOG_ERR(strerror(errno));
                    continue;
                }

                set_fd_noblk(cli_sock);

                println("[sock:", cli_sock, ']', "[client:", inet_ntoa(cli_addr.sin_addr), ':', ntohs(cli_addr.sin_port), ']');

                srv_event.events = EPOLLIN | EPOLLET;
                srv_event.data.fd = cli_sock;

                epoll_ctl(ep_fd, EPOLL_CTL_ADD, cli_sock, &srv_event);
            }
            else {
                char * _ptr_r = read_buf;

                int r_len = recv(_sock, read_buf, READ_BUF_SIZE, 0);

                if (r_len <= 0) {
                    close_fds(_sock);
                    epoll_ctl(ep_fd, EPOLL_CTL_DEL, _sock, nullptr);
                    println("closed fd: ", _sock);
                    continue;
                }
                else {
                    _ptr_r += r_len;
                    while (r_len = recv(_sock, _ptr_r, READ_BUF_SIZE, 0) > 0) {
                        _ptr_r += r_len;
                    }
                }

                std::string r_msg(read_buf, _ptr_r);
                println("[recv]<", r_msg, '>');

                std::pair<char, std::string> _ret = base.handle_f(r_msg);
                std::string w_msg;
                w_msg.push_back(_ret.first >= 0 ? 'S' : 'E');
                w_msg.append(_ret.second);
                send(_sock, w_msg.c_str(), w_msg.size(), 0);

                println("[send]<", w_msg, '>');

                if (r_msg == "#exit") {
                    close_fds(_sock);
                    epoll_ctl(ep_fd, EPOLL_CTL_DEL, _sock, nullptr);
                    println("closed fd: ", _sock);
                    continue;
                }
            }
        }
    }

    return 0;
}
#endif

}

